package priv.abadstring.study.haoke.im.websocket;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.bson.types.ObjectId;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import priv.abadstring.study.haoke.im.dao.MessageDAO;

/**
 * RocketMQListenerImplToSendMessage
 *
 * @author aBadString
 * @date 2021-08-05 09:47:40
 */
@Component
@RocketMQMessageListener(
        topic = "haoke-im",
        selectorExpression = "SEND_MESSAGE",
        consumerGroup = "haoke-im-consumer",
        messageModel = MessageModel.BROADCASTING
)
public class RocketMQListenerImplToSendMessage implements RocketMQListener<String> {
    @Autowired
    private MessageDAO messageDAO;

    private static final ObjectMapper MAPPER = new ObjectMapper();

    @Override
    public void onMessage(String message) {
        try {
            JsonNode jsonNode = MAPPER.readTree(message);
            System.out.println(message);
            Long toId = jsonNode.get("to").get("id").longValue();
            // 判断to用户是否在线
            WebSocketSession toSession = MessageHandler.SESSIONS.get(toId);
            if (toSession != null && toSession.isOpen()) {
                toSession.sendMessage(new TextMessage(message));
                // 更新消息状态为已读
                this.messageDAO.updateMessageState(new ObjectId(jsonNode.get("id").asText()), 2);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
